package com.iflytek.uoamp.agent.monitor.sender.biz.quick;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.log4j.Logger;

import com.iflytek.uoamp.agent.common.util.LogHelper;
import com.iflytek.uoamp.agent.core.sender.monitorsender.monitorinterface.IItem;
import com.iflytek.uoamp.agent.monitor.mcagent.SendClient;
import com.iflytek.uoamp.agent.monitor.sender.biz.AbsSender;
import com.iflytek.uoamp.agent.monitor.sender.config.Config;
import com.iflytek.uoamp.agent.monitor.sender.interfaces.ISender;
import com.iflytek.uoamp.agent.monitor.sender.interfaces.ISenderFactory;
import com.iflytek.uoamp.agent.monitor.sender.loadbalance.LoadbalanceManager;

public class QuickSender extends AbsSender {
	private static final byte QUEUE_IOTIMEOUT = 5;
	private final QuickQueue<IItem> sendQueue;
	private final ExecutorService sendThreadPool;
	private ISender clientSender = SendClient.getInstance();
	private void sendItem() {
		IItem cur = null;
//		ConnectionPoolManager connectionPoolManager = null;
//		ISender clientSender = null;
		while (true) {
			try {
				cur = sendQueue.poll(QUEUE_IOTIMEOUT * 12);
				if (null != cur) {
//					connectionPoolManager = lbManager.getNextConnectManager();
//					if (null == connectionPoolManager) {
//						continue;
//					}
					long begin = System.currentTimeMillis();

					try {
//						clientSender = connectionPoolManager.checkout();
						if (null == clientSender) {
							Logger.getRootLogger().error("QuickSender sendItem request client failed.");
							continue;
						}
						clientSender.send(cur);
					} catch (Exception e) {
						Logger.getRootLogger().error("QuickSender sendItem failed.", e);
					} finally {
//						if (null != clientSender) {
//							connectionPoolManager.checkIn(clientSender);
//						}
					}
					long end = System.currentTimeMillis();
					LogHelper.info(Config.getLog(),"Send Event in thread use time:" + (end - begin));
				}
			} catch (Exception e) {
				Logger.getRootLogger().error(ExceptionUtils.getStackTrace(e),e);
			}
		}
	}
	
	@SuppressWarnings("rawtypes")
	public QuickSender(LoadbalanceManager lbManager,ISenderFactory factory, Config config) {
		super(lbManager,factory, config);
		final AtomicLong threadCounter = new AtomicLong(0);
		this.sendQueue = new QuickQueue<IItem>(config.getSendQueueSize());

		this.sendThreadPool = Executors.newFixedThreadPool(config.getConnectPoolSize(),  new ThreadFactory() {
			@Override
			public Thread newThread(Runnable r) {
				Thread t = new Thread(r);
				t.setName("Quick Sender thread pool - " + String.valueOf(threadCounter.incrementAndGet()));
				Logger.getRootLogger().info("Make a new Safe Sender thread : " + t.getName());
				return t;
			}
		});
		
		for (int i = 0; i < config.getConnectPoolSize(); i++) {
			this.sendThreadPool.execute(new Runnable() {
				@Override
				public void run() {
					sendItem();
				}
			});
		}
	}

	@Override
	public Object send(IItem item) {
		try {
			return sendQueue.offer(item, QUEUE_IOTIMEOUT);
		} catch (Exception e) {
			Logger.getRootLogger().error(ExceptionUtils.getStackTrace(e),e);
		}
		Logger.getRootLogger().error("QuickSender send Item timeout : sendQueue.offer exceed 5 seconds.");
		return false;
	}

	@Override
	public Object sendBatch(List<IItem> item) {
		boolean result = true;
		for (IItem cur : item) {
			result = (Boolean)send(cur);
			if (result) {
				break;
			} 
		}
		return result;
	}
	
}
